use crate::func::RedisCache;
use crate::{Column, GroupCalc, LowCardinality};
use chrono::{DateTime, Datelike};
use chrono_tz::Tz;
use clickhouse_rs::types::{Complex, Decimal, Row, SqlType};
use clickhouse_rs::ClientHandle;
use serde_json::Value;
use shq_common::prelude::report::{DrillDownData, ReportDataResponse, ReportGroup, Unit};
use shq_common::prelude::{get_ch_pool, DataType, OrderItems};
use std::borrow::Cow;
pub(crate) struct TableReader<'a> {
    tmp_table_name: &'a str,
    leafs: &'a [&'a Column],
    rept_unit: Unit,
}
impl<'a> TableReader<'a> {
    pub(crate) fn new(
        tmp_table_name: &'a str,
        columns: &'a Vec<&'a Column>,
        rept_unit: Unit,
    ) -> Self {
        TableReader {
            tmp_table_name,
            leafs: columns,
            rept_unit,
        }
    }
    pub(crate) async fn get_page_data(
        &self,
        where_cause: &str,
        orday_by: &str,
        limit: u16,
        offset: u64,
    ) -> anyhow::Result<ReportDataResponse> {
        let mut rc = RedisCache::new(self.tmp_table_name)?;
        if rc.is_exist()? {
            let row_datas = rc.get()?;
            let rlen = row_datas.len();
            return Ok(ReportDataResponse::new(row_datas, rlen as u64));
        }
        let mut client = get_ch_pool().get_handle().await?;
        let sql_str = format!(
            "select count(*) from {} {}",
            self.tmp_table_name, where_cause,
        );
        let block = client.query(sql_str).fetch_all().await?;
        let mut count = 0u64;
        if let Some(row) = block.rows().next() {
            count = row.get(0)?;
        }
        let col_names: Vec<&str> = self
            .leafs
            .iter()
            .map(|c| c.name.as_deref().unwrap())
            .collect();
        let col_str = col_names.join(",");
        let mut sql_str = format!(
            "select {} from {} {} {orday_by}",
            col_str, self.tmp_table_name, where_cause,
        );
        if limit > 0 {
            sql_str.push_str(format!(" limit {} offset {}", limit, offset).as_str());
        }
        let (mut row_datas, _) =
            query_to_list(&mut client, &sql_str, self.leafs, &self.rept_unit).await?;
        let psize = limit as u64;
        let has_total = self.leafs.iter().any(|l| l.col_total_calc.is_some());
        if has_total && (limit == 0 || offset + psize >= count) && count > 0 {
            let trow = get_total_row(
                &mut client,
                self.leafs,
                self.tmp_table_name,
                where_cause,
                &self.rept_unit,
            )
            .await?;
            if let Some(trow) = trow {
                row_datas.push(trow);
            }
        }
        if has_total {
            count += 1;
        }
        Ok(ReportDataResponse::new(row_datas, count))
    }
    pub(crate) async fn drill_down(
        &self,
        requs: Vec<DrillDownRequ<'a>>,
        orders: Option<&OrderItems>,
    ) -> anyhow::Result<Vec<DrillDownData>> {
        let mut res: Vec<DrillDownData> = Vec::new();
        let mut client = get_ch_pool().get_handle().await?;
        for requ in requs.into_iter() {
            let col_names: Vec<&str> = self
                .leafs
                .iter()
                .map(|c| {
                    let n = c.name.as_deref().unwrap();
                    if n == requ.col_name_map.0 {
                        requ.col_name_map.1.as_ref()
                    } else {
                        n
                    }
                })
                .collect();
            for val in requ.gourp_values.iter() {
                let where_cause = format!(
                    "where gp_index={} and {}={}",
                    requ.group_index + 1,
                    requ.group_by_col_name,
                    val
                );
                let def_order = requ.group_by_col_name;
                let order_by = parse_order(orders, def_order);
                let col_str = col_names.join(",");
                let sql_str = format!(
                    "select {} from {} {} {order_by}",
                    col_str, self.tmp_table_name, where_cause,
                );
                let (data, _) =
                    query_to_list(&mut client, &sql_str, self.leafs, &self.rept_unit).await?;
                res.push(DrillDownData {
                    group_index: requ.group_index,
                    by_col: requ.group_by_col_name.to_string(),
                    by_value: val.clone(),
                    data,
                });
            }
        }
        Ok(res)
    }
}
pub(crate) async fn query_to_list(
    client: &mut ClientHandle,
    sql_str: &str,
    leafs: &[&Column],
    rept_unit: &Unit,
) -> anyhow::Result<(Vec<Vec<Value>>, Vec<SqlType>)> {
    let block = client.query(sql_str).fetch_all().await?;
    let cols = block.columns();
    let cols: Vec<SqlType> = cols.iter().map(|col| col.sql_type().clone()).collect();
    if cols.len() != leafs.len() {
        anyhow::bail!(
            "查询结果有{}列数据,而配置中有{}列,两者不一致,请检查自定义函数",
            cols.len(),
            leafs.len()
        )
    }
    let mut row_datas: Vec<Vec<Value>> = Vec::new();
    for row in block.rows() {
        let mut row_data: Vec<Value> = Vec::new();
        for (i, col) in cols.iter().enumerate() {
            let l = leafs[i];
            let fmt = l.format.as_deref();
            let unit = if l.title.is_none() {
                1
            } else if let Some(unit) = l.unit {
                unit.to_u32()
            } else {
                rept_unit.to_u32()
            };
            let jv = to_json_value(&row, col, i, fmt, unit)?;
            row_data.push(jv);
        }
        row_datas.push(row_data);
    }
    Ok((row_datas, cols))
}
pub(crate) async fn get_total_row(
    client: &mut ClientHandle,
    leafs: &[&Column],
    tmp_table_name: &str,
    where_cause: &str,
    rept_unit: &Unit,
) -> anyhow::Result<Option<Vec<Value>>> {
    let cols_iter = leafs.iter().filter(|l| l.col_total_calc.is_some());
    let mut col_exps: Vec<String> = Vec::new();
    let mut agg_cols: Vec<&Column> = Vec::new();
    for col in cols_iter {
        let gcalc = col.col_total_calc.as_ref().unwrap();
        let col_name = col.name.as_ref().unwrap();
        let exp = if let GroupCalc::Str(s) = gcalc {
            let mut s = s.to_owned();
            if s.contains("#PREV_COL") {
                if let Some(prev_col) = agg_cols.last() {
                    let idx = prev_col
                        .name
                        .as_ref()
                        .map(|n| n.as_str().rfind('_'))
                        .flatten();
                    if let Some(idx) = idx {
                        let prev_name = &prev_col.name.as_ref().unwrap()[..idx];
                        s = s.replace("#PREV_COL", prev_name);
                    }
                }
            }
            s
        } else {
            format!("{gcalc}({col_name})")
        };
        col_exps.push(exp);
        agg_cols.push(*col);
    }
    if agg_cols.is_empty() {
        return Ok(None);
    }
    let sel_cols = col_exps.join(",");
    let sql_str = format!("select {sel_cols} from {tmp_table_name} {where_cause}");
    let block = client.query(sql_str).fetch_all().await?;
    let cols = block.columns();
    let col_types: Vec<SqlType> = cols.iter().map(|col| col.sql_type().clone()).collect();
    let mut agg_vals: Vec<Value> = Vec::new();
    let data_row = block.rows().next().unwrap();
    for (i, st) in col_types.iter().enumerate() {
        let fmt = None;
        let col = agg_cols[i];
        let unit = if let Some(u) = col.unit {
            u.to_u32()
        } else {
            rept_unit.to_u32()
        };
        let jv = to_json_value(&data_row, st, i, fmt, unit)?;
        agg_vals.push(jv);
    }
    let mut res: Vec<Value> = Vec::new();
    let mut is_set = false;
    for leaf in leafs.iter() {
        match leaf.data_type {
            DataType::Bool => res.push(false.into()),
            DataType::Int | DataType::Float => {
                let cname = leaf.name.as_ref().unwrap();
                let acol = agg_cols.iter().find(|c| c.name.as_ref().unwrap() == cname);
                if let Some(acol) = acol {
                    let idx = agg_cols.iter().position(|c| c == acol).unwrap();
                    res.push(agg_vals[idx].clone());
                } else {
                    res.push(0.into());
                }
            }
            DataType::String => {
                if !is_set && leaf.display && leaf.title.is_some() {
                    res.push("合计".into());
                    is_set = true;
                } else {
                    res.push("".into());
                }
            }
            _ => res.push("".into()),
        }
    }
    Ok(Some(res))
}
pub(crate) struct DrillDownRequ<'a> {
    pub(crate) group_index: u8,
    pub(crate) group_by_col_name: &'a str,
    pub(crate) gourp_values: Vec<Value>,
    pub(crate) col_name_map: (&'a str, Cow<'a, str>),
}
impl<'a> DrillDownRequ<'a> {
    pub(crate) fn from_vec(
        drill_down: Vec<(u8, Value)>,
        groups: &'a Vec<ReportGroup>,
    ) -> Vec<Self> {
        let mut drill_down_requs: Vec<DrillDownRequ<'a>> = vec![];
        let glen = groups.len();
        let min_gidx = drill_down.iter().map(|(g, _)| *g).min().unwrap_or(0);
        for (g, v) in drill_down.into_iter() {
            let ddr = drill_down_requs.iter_mut().find(|d| d.group_index == g);
            if let Some(ddr) = ddr {
                ddr.gourp_values.push(v);
            } else {
                let idx = g as usize;
                if idx >= glen {
                    continue;
                }
                let rg1 = &groups[idx];
                let rg2 = &groups[idx + 1];
                let rg2_len = rg2.name_cols.len();
                let cv = if rg2_len == 1 {
                    Cow::Borrowed(rg2.name_cols[0].as_str())
                } else {
                    let mut col_exp: String = "concat(".to_string();
                    let n = idx + 1 - min_gidx as usize;
                    let prev = vec!["  "; n].join("");
                    col_exp.push_str("'");
                    col_exp.push_str(prev.as_str());
                    col_exp.push_str("',");
                    for (i, nc) in rg2.name_cols.iter().enumerate() {
                        col_exp.push_str("trimLeft(");
                        col_exp.push_str(nc.as_str());
                        col_exp.push(')');
                        if i < rg2_len - 1 {
                            col_exp.push_str(",");
                            col_exp.push_str("'-',");
                        }
                    }
                    col_exp.push_str(")");
                    Cow::Owned(col_exp)
                };
                let ddr = DrillDownRequ {
                    group_index: g,
                    group_by_col_name: rg1.by.as_str(),
                    gourp_values: vec![v],
                    col_name_map: (rg1.name_cols[0].as_str(), cv),
                };
                drill_down_requs.push(ddr);
            }
        }
        drill_down_requs
    }
}
pub(crate) fn to_json_value(
    row: &Row<'_, Complex>,
    st: &SqlType,
    i: usize,
    fmt: Option<&str>,
    unit: u32,
) -> anyhow::Result<serde_json::Value> {
    match st {
        SqlType::Bool => {
            let v: bool = row.get(i)?;
            Ok(serde_json::Value::Bool(v))
        }
        SqlType::UInt8 => {
            let v: u8 = row.get(i)?;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::UInt16 => {
            let v: u16 = row.get(i)?;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::UInt32 => {
            let mut v: u32 = row.get(i)?;
            v = v / unit;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::UInt64 => {
            let mut v: u64 = row.get(i)?;
            v = v / unit as u64;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::Int8 => {
            let v: i8 = row.get(i)?;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::Int16 => {
            let v: i16 = row.get(i)?;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::Int32 => {
            let mut v: i32 = row.get(i)?;
            v = v / unit as i32;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::Int64 => {
            let mut v: i64 = row.get(i)?;
            v = v / unit as i64;
            Ok(serde_json::Value::Number(serde_json::Number::from(v)))
        }
        SqlType::String => {
            let v: String = row.get(i)?;
            Ok(serde_json::Value::String(v))
        }
        SqlType::FixedString(_) => {
            let v: String = row.get(i)?;
            Ok(serde_json::Value::String(v))
        }
        SqlType::Float32 => {
            let mut v: f32 = row.get(i)?;
            v = v / unit as f32;
            if v.is_nan() {
                v = 0.0;
            }
            Ok(serde_json::Value::Number(
                serde_json::Number::from_f64(v as f64).unwrap(),
            ))
        }
        SqlType::Float64 => {
            let mut v: f64 = row.get(i)?;
            v = v / unit as f64;
            if v.is_nan() {
                v = 0.0;
            }
            Ok(serde_json::Value::Number(
                if let Some(jv) = serde_json::Number::from_f64(v) {
                    jv
                } else {
                    serde_json::Number::from_f64(0.0).unwrap()
                },
            ))
        }
        SqlType::Date => {
            let v: chrono::NaiveDate = row.get(i)?;
            let fmt = if let Some(fmt) = fmt { fmt } else { "%Y-%m-%d" };
            if v.year() >= 2149 || v.year() <= 1970 {
                Ok(serde_json::Value::String("".to_string()))
            } else {
                Ok(serde_json::Value::String(v.format(fmt).to_string()))
            }
        }
        SqlType::DateTime(_) => {
            let v: DateTime<Tz> = row.get(i)?;
            let fmt = if let Some(fmt) = fmt {
                fmt
            } else {
                "%Y-%m-%d %H:%M:%S"
            };
            if v.year() >= 2149 || v.year() <= 1970 {
                Ok(serde_json::Value::String("".to_string()))
            } else {
                Ok(serde_json::Value::String(v.format(fmt).to_string()))
            }
        }
        SqlType::LowCardinality(sql_type) => to_json_value(row, sql_type, i, fmt, unit),
        SqlType::Decimal(_p, s) => {
            let v: Decimal = row.get(i)?;
            let iv: i64 = v.internal();
            let v = iv as f64 / 10i64.pow(*s as u32) as f64;
            Ok(serde_json::Value::Number(
                serde_json::Number::from_f64(v).unwrap(),
            ))
        }
        _ => anyhow::bail!("不支持的数据类型:{}", st),
    }
}

pub(crate) fn parse_order(orders: Option<&OrderItems>, default: &str) -> String {
    let ostr = if let Some(o) = orders {
        let os = o.to_order_string();
        os
    } else {
        default.to_string()
    };
    if ostr.is_empty() {
        "".to_string()
    } else {
        format!("order by {}", ostr)
    }
}
pub(crate) async fn query_string_value(sql: &str) -> anyhow::Result<String> {
    let mut client = get_ch_pool().get_handle().await?;
    let block = client.query(sql).fetch_all().await?;
    let res = if let Some(row) = block.rows().next() {
        let val: Result<String, clickhouse_rs::errors::Error> = row.get(0);
        if let Ok(val) = val {
            val
        } else {
            let val: Option<String> = row.get(0)?;
            val.map(|s| s.to_string()).unwrap_or_else(|| "".to_string())
        }
    } else {
        "".to_string()
    };
    Ok(res)
}

pub(crate) async fn get_low_cardinality_col_options(
    tmp_table_name: &str,
    col_name: &str,
    lowc: &LowCardinality,
) -> anyhow::Result<Vec<String>> {
    let mut client = get_ch_pool().get_handle().await?;
    let oexp= match lowc {
        LowCardinality::Asc => format!("{} asc", col_name),
        LowCardinality::Desc => format!("{} desc", col_name),
        LowCardinality::Tops(tops) => {
            let mut exp=format!("case {}", col_name);
            let mut i=0_usize;
            for s in tops.iter() {
                exp.push_str(format!(" when '{}' then {}", s, i).as_str());
                i+=1;
            }
            exp.push_str(" else ");
            exp.push_str(format!("{}", i).as_str());
            exp.push_str(" end");
            exp
        }
    };
    let sql_str =
        format!("select distinct toString({col_name}) from {tmp_table_name} order by {oexp}",);
    let block = client.query(sql_str).fetch_all().await?;
    let mut options: Vec<String> = Vec::new();
    for row in block.rows() {
        let val = row.get::<String, _>(0)?;
        options.push(val);
    }
    Ok(options)
}
